﻿#region License
/* ---------------------------------------------------------------- *
 *
 * This file is part of the Xcoordination Application Space
 * ("XcoAppSpace") http://www.xcoordination.com
 *
 * Copyright (C) 2009 Xcoordination http://www.xcoordination.com
 *
 * XcoAppSpace is free software; you can redistribute it and/or
 * modify it under the terms of version 2 of the GNU General
 * Public License as published by the Free Software Foundation.
 *
 * XcoAppSpace is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see http://www.gnu.org/licenses/
 * or write to the Free Software Foundation, Inc., 51 Franklin
 * Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 * ---------------------------------------------------------------- */
#endregion

using System;
using System.Collections.Generic;
using XcoAppSpaces.Contracts.Exceptions;
using XcoAppSpaces.Contracts.Transport;
using System.Threading;
using XcoAppSpaces.Contracts.Logging;
using System.Messaging;

namespace XcoAppSpaces.Transport.MSMQ
{
	/// <summary>
	/// Communication Service that send/receives messages to/from remote spaces via
	/// MSMQ. The service uses one queue for reading messages.
	/// </summary>
	public class XcoMSMQTransportService : IXcoTransportService
	{
		/// <summary>
		/// Event that is raised when a message is received.
		/// </summary>
		public event MessageReceivedDelegate OnMessageReceived;
		

        private string queueName;
		private bool running;
		
		private IXcoLogger log;
		private MSMQServer server;
        private readonly Dictionary<string, MSMQSend> activeSenders = new Dictionary<string, MSMQSend>();

		private const int timeoutCheckInterval = 30000; //check for connection timeouts every 30 seconds
		private const int connectionTimeout = 60; //60 seconds connection timeout
		private Timer timeoutCheckTimer; //the timer that triggers the timeout checks
				
        private string queuePath = "notInit";

		/// <summary>
		/// Remove Queue after use if true
		/// </summary>
		public bool RemoveQueueOnStop {get;set;}

		/// <summary>
        /// Creates a new XcoMSMQTransportService.
		/// </summary>
		public XcoMSMQTransportService() : this("", false)
		{}

		/// <summary>
        /// Creates a new XcoMSMQTransportService.
		/// </summary>
        /// <param name="queueName">The queueName where the service should read from.</param>
        public XcoMSMQTransportService(string queueName) : this(queueName, false)
		{}

        /// <summary>
        /// Creates a new XcoMSMQTransportService.
        /// </summary>
        /// <param name="queueName">The queueName where the service should read from.</param>
        /// <param name="removeQueueOnStop">If true the queue will be removed upon disposal of communication service</param>
        public XcoMSMQTransportService(string queueName, bool removeQueueOnStop)
        {
            RemoveQueueOnStop = removeQueueOnStop;
            this.queueName = queueName;
			this.log = XcoLoggerFactory.CreateInstance(this.GetType());
        }
        
        /// <summary>
		/// The message queue on which the service reads from.
		/// </summary>
        public string QueueName
		{
            get { return queueName; }
			set
			{
				if (!running)
                    queueName = value;
				else
					throw new XcoCommunicationException("Queue-name cannot be set while the communication service is running.");
			}
		}

		

        /// <summary>
        /// Delete queue 
        /// </summary>
        public void DeleteQueue()
        {
            if (!running)
            {
                if (MessageQueue.Exists(queueName))
                    MessageQueue.Delete(queueName);
            }
            else
                throw new XcoCommunicationException("Queue cannot be deleted while the communication service is running.");

        }

		/// <summary>
		/// Initializes the transport service.
		/// </summary>
		/// <param name="serviceRegistry">Service registry containing other services of the space.</param>
		public void Initialize(XcoAppSpaces.Contracts.Service.IXcoServiceRegistry serviceRegistry)
		{
		}

		/// <summary>
		/// The start priority of the service.
		/// </summary>
		public int StartPriority
		{
			get { return 5; }
		}

		/// <summary>
		/// Starts the communication service.
		/// </summary>
		public void Start()
		{
            if (server == null)
            {
                log.Info("Starting XcoMSMQTransportService on queue " + queueName);
                
                //start server
                server = new MSMQServer(queueName, this, log);

                queuePath = server.QueuePath;
                running = true;

                //start timeout check timer
                timeoutCheckTimer = new Timer(CheckConnectionTimeouts, null, timeoutCheckInterval, timeoutCheckInterval);
            }
		}

		/// <summary>
		/// Stops the service and closes all open connections.
		/// </summary>
		public void Stop()
		{
			if (running)
			{
				running = false;
				timeoutCheckTimer.Dispose();
				ReleaseConnections();
				server.Stop();
				
				if (RemoveQueueOnStop)
				{
					try
					{
						DeleteQueue();
					}
					catch (Exception ex)
					{
						log.Error("Error during dispose - remove queue from system", ex);
					}
				}
			}
		}
		
		/// <summary>
		/// Releases all currently active connections.
		/// </summary>
		private void ReleaseConnections()
		{
			lock (activeSenders)
			{
				foreach (MSMQSend client in activeSenders.Values)
				{
					client.Close();
				}
				activeSenders.Clear();
			}
		}

		/// <summary>
		/// Sends a message to a remote space.
		/// </summary>
		/// <param name="msg">The messages that should be sent.</param>
		/// <param name="remoteAddress">The address where the message should be sent</param>
		public void Send(XcoMessage msg, string remoteAddress)
		{
			if (!running)
				msg.RaiseTransferFailedEvent(new XcoCommunicationException("Server for remote communication has not been started!"));
			bool newConnection = false;
			try
			{
				MSMQSend sender;
				lock (activeSenders)
				{
					//find remote sender or create new one, if no sender to this remote address exists
					if (activeSenders.ContainsKey(remoteAddress))
						sender = activeSenders[remoteAddress];
					else
					{
						sender = new MSMQSend(remoteAddress);
						activeSenders.Add(remoteAddress, sender);
						newConnection = true;
					}
				}

				log.Debug("Sending message to " + remoteAddress);
				//before sending: set remote address of the message to own address, so remote space 
				//knows where the message comes from
				//msg.RemoteAddress = queuePath;
				//send message
				sender.Send(msg, remoteAddress, this.LocalAddress);
				msg.RaiseTransferSuccessfulEvent();
			}
			catch (Exception ex)
			{
				//retry opening a new connection if the present connection is not new
				if (!newConnection)
				{
					try
					{
						log.Warn("Error sending message to address: " + remoteAddress + " using the currently open connection. Trying to open new connection.");
						MSMQSend sender;
						lock (activeSenders)
						{
							sender = new MSMQSend(remoteAddress);
							activeSenders[remoteAddress] = sender;
						}
						//msg.RemoteAddress = queuePath;
						//send message
						sender.Send(msg, remoteAddress, this.LocalAddress);
						msg.RaiseTransferSuccessfulEvent();
					}
					catch (Exception ex2)
					{
						log.Error("Error sending message to address: " + remoteAddress, ex2);
						//msg.RemoteAddress = remoteAddress;
						msg.RaiseTransferFailedEvent(new XcoCommunicationException("Error sending message to address \"" + remoteAddress + "\" : " + ex2.Message, ex2));
					}
				}
				else
				{
					log.Error("Error sending message to address: " + remoteAddress, ex);
					//msg.RemoteAddress = remoteAddress;
					msg.RaiseTransferFailedEvent(new XcoCommunicationException("Error sending message to address \"" + remoteAddress + "\" : " + ex.Message, ex));
				}
			}
		}

		/// <summary>
		/// Checks if any TCPClients have run into a timeout (not sent any message for a 
		/// defined amount of time), and removes these clients and closes their connections.
		/// </summary>
		/// <param name="timerInfo">Field for additional info from the timer that starts the method (is ignored because not additional info needed).</param>
		private void CheckConnectionTimeouts(object timerInfo)
		{
			DateTime checkDate = DateTime.Now.AddSeconds(-connectionTimeout);

			//keep access time to the remoteSenders Dictionary as short as possible
			List<MSMQSend> timedOut = new List<MSMQSend>();
			List<string> timedOutKeys = new List<string>();
			lock (activeSenders)
			{
				foreach (String address in activeSenders.Keys)
				{
					MSMQSend sender = activeSenders[address];
					if (sender.LastSendTime <= checkDate)
					{
						timedOut.Add(sender);
						timedOutKeys.Add(address);
						log.Debug("Sender timed out for remote kernel: " + address);
					}
				}
				//remove RemoteSenders from the list
				foreach (String key in timedOutKeys)
				{
					activeSenders.Remove(key);
				}
			}
			//disconnect senders that have timed out
			foreach (MSMQSend sender in timedOut)
			{
				sender.Close();
			}
		}

		internal void RaiseMessageReceivedEvent(XcoMessage msg, string remoteAddress)
		{
			OnMessageReceived(msg, remoteAddress, this);
		}

        #region IDisposable Members

		/// <summary>
		/// Stops the service and closes all open connections.
		/// </summary>
		public void Dispose()
        {
            this.Stop();			
        }

        #endregion

        #region IXcoTransportService Members

        /// <summary>
        /// Returns the MSMQ name
        /// </summary>
        public string LocalAddress
        {
            get { return queuePath; }
        }

        #endregion
    }
}
